package spark.person

import java.time.LocalDateTime
import java.util

import com.sun.jersey.core.util.Base64
import org.apache.hadoop.hbase.TableName
import org.apache.hadoop.hbase.client.Put
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.SparkConf
import org.apache.spark.sql.{Row, SparkSession}
import utils.HBaseUtil

/**
  * Created by zhangbn on 2018/10/16.
  */
object SearchInPersonBase {

  def main(args: Array[String]): Unit = {

    System.setProperty("hadoop.home.dir", "E:\\hadoop-common-2.7.3-bin-master")
    System.setProperty("HADOOP_USER_NAME", "admin")

    val startTime = System.currentTimeMillis
    println(s"[${LocalDateTime.now()}]  SearchInPersonBase作业启动.......................  now_ms=${startTime}")

    val sparkConf = new SparkConf()
      .setAppName("SearchInPersonBase")
      .setMaster("local[*]")

    sparkConf.set("spark.sql.crossJoin.enabled", "true")

    val spark = SparkSession
      .builder()
      .config(sparkConf)
      .enableHiveSupport()
      .getOrCreate()

    spark.sqlContext.sql("use ods")
    val oriDF = spark.sql("select * from t_person_passinfo where dt='180618' limit 5")

//    println(oriDF.rdd.count())
//    println(oriDF.rdd.partitions.size)
    oriDF.rdd.foreach(println(_))
//    val featureList = oriDF.select("feature")
//      .map(row => {
//        val str = row.get(0).asInstanceOf[String]
//        Base64.decode(str)
//      }).collectAsList()
//
//    val passIdList = oriDF.select("pass_id")
//      .map(row => {
//        row.get(0).asInstanceOf[String]
//      }).collectAsList()

//    val featureListBC = spark.sparkContext.broadcast(featureList)
//    val passIdListBC = spark.sparkContext.broadcast(passIdList)

    val hbase = new HBaseUtil
    val baseVoList = hbase.getBaseTableColumn("personBase")
    val hFeatureList = hbase.getFeatureList(baseVoList)

    //val testTbList: util.List[Put] = new util.ArrayList[Put]

    oriDF.rdd.foreach(row => {
      val pass_id = row.getAs[String]("pass_id")
      val var1 = Base64.decode(row.getAs[String]("feature"))
      val doubleArr = featureCompByByte(var1, hFeatureList)
      var max = doubleArr(0)
      var idx = 0
      for ( i <- 1 to (doubleArr.length - 1) ) {
        if (doubleArr(i) > max){
            max = doubleArr(i)
            idx = i
          }
      }
      val h =  new HBaseUtil()
      if(max>=0.8){
        val uuid=baseVoList.get(idx).getUuid().toString
//        val put = new Put(Bytes.toBytes(uuid))
//        put.addColumn("f1".getBytes, "pass_id".getBytes, Bytes.toBytes(pass_id))
//        HBaseUtil.connection.getTable(TableName.valueOf("180618".getBytes())).put(put)
        h.putDataH("180618",uuid,"f1","pass_id",pass_id)
      }else{
        //println(s"新增记录：${uuid}     ${pass_id}")
        h.putDataH("180618",pass_id,"f1","pass_id",pass_id)
      }



    })

//    oriDF.rdd.foreachPartition(partitionsFun)

    val endTime = System.currentTimeMillis
    println(s"[${LocalDateTime.now()}]  SearchInPersonBase作业结束.......................  now_ms=${endTime}")

  }


  def insertTestTB(iter: Iterator[Row]) = {
    val lists = new util.ArrayList[Put]
    while (iter.hasNext) {
      val next = iter.next()
      val rowkey = next.getAs[Integer]("uuid")
      val put = new Put(Bytes.toBytes(rowkey))
      val cf = "info"
      put.addColumn(cf.getBytes, "pass_id".getBytes, next.getAs[String]("pass_id").getBytes)
      put.addColumn(cf.getBytes, "pass_time".getBytes, next.getAs[String]("pass_time").getBytes)
      put.addColumn(cf.getBytes, "camera_id".getBytes, next.getAs[String]("camera_id").getBytes)
      put.addColumn(cf.getBytes, "camera_code".getBytes, Bytes.toBytes(next.getAs[Long]("camera_code")))
      put.addColumn(cf.getBytes, "device_id".getBytes, next.getAs[String]("device_id").getBytes)
      put.addColumn(cf.getBytes, "server_id".getBytes, next.getAs[String]("server_id").getBytes)
      put.addColumn(cf.getBytes, "face_img_url".getBytes, next.getAs[String]("face_img_url").getBytes)
      put.addColumn(cf.getBytes, "environ_img_url".getBytes, next.getAs[String]("environ_img_url").getBytes)
      put.addColumn(cf.getBytes, "quality_score".getBytes, Bytes.toBytes(next.getAs[Double]("quality_score")))
      put.addColumn(cf.getBytes, "age".getBytes, Bytes.toBytes(next.getAs[Int]("age")))
      put.addColumn(cf.getBytes, "gender".getBytes, Bytes.toBytes(next.getAs[Int]("gender")))
      put.addColumn(cf.getBytes, "attractive".getBytes, Bytes.toBytes(next.getAs[Int]("attractive")))
      put.addColumn(cf.getBytes, "eyeglass".getBytes, Bytes.toBytes(next.getAs[Int]("eyeglass")))
      put.addColumn(cf.getBytes, "sunglass".getBytes, Bytes.toBytes(next.getAs[Int]("sunglass")))
      put.addColumn(cf.getBytes, "smile".getBytes, Bytes.toBytes(next.getAs[Int]("smile")))
      put.addColumn(cf.getBytes, "mask".getBytes, Bytes.toBytes(next.getAs[Int]("mask")))
      put.addColumn(cf.getBytes, "race".getBytes, Bytes.toBytes(next.getAs[Int]("race")))
      put.addColumn(cf.getBytes, "eyeopen".getBytes, Bytes.toBytes(next.getAs[Int]("eyeopen")))
      put.addColumn(cf.getBytes, "mouthopen".getBytes, Bytes.toBytes(next.getAs[Int]("mouthopen")))
      put.addColumn(cf.getBytes, "beard".getBytes, Bytes.toBytes(next.getAs[Int]("beard")))
      put.addColumn(cf.getBytes, "feature".getBytes, next.getAs[String]("feature").getBytes)
      put.addColumn(cf.getBytes, "create_time".getBytes, next.getAs[String]("create_time").getBytes)
      lists.add(put)
    }
    val tableName = "personBase"
    val table = HBaseUtil.connection.getTable(TableName.valueOf(tableName.getBytes()))
    try
      table.put(lists)
    catch {
      case ex: Exception => println("save to HBase ERROR：" + ex.printStackTrace())
    } finally table.close()
  }

  def featureCompByByte(var1: Array[Byte], var2: util.List[Array[Byte]]): Array[Double] = {

    val doubleArr = new Array[Double](5)
    for ( i <- 0 to (doubleArr.length - 1) ){
      doubleArr(i)=Math.random()
    }
    doubleArr
//    Array(0.1, 0.2, 0.3, 0.4, 0.5)
  }

}

